package my

import akka.actor.{Actor, ActorLogging, ActorRef, Cancellable, Props}
import akka.cluster.routing.{ClusterRouterPool, ClusterRouterPoolSettings}
import akka.io.Udp.SO.Broadcast
import akka.routing.{BalancingPool, BroadcastPool}
import my.MyMaster.{Enlist, NextTask, StartJob}

import scala.concurrent.duration._

/**
  * Created by shichen on 2017/8/3.
  */
object MyMaster {
  def props() = Props(new MyMaster)

  case class StartJob(name: String,text: List[String])
  case class Enlist(worker: ActorRef)
  case object NextTask
  case class WorkResult(map: Map[String,Int])

  case object MergeResult
}

trait CreateWorkerRouter { this: Actor =>
  def createWorkerRouter: ActorRef = {
    context.actorOf(
      ClusterRouterPool(BroadcastPool(10),
        ClusterRouterPoolSettings(
          totalInstances = 50,
          maxInstancesPerNode = 5,
          allowLocalRoutees = false,
          useRole = None
        )
      ).props(Props[MyWorker]),name = "worker-router"
    )
  }
}

class MyMaster extends Actor with ActorLogging with CreateWorkerRouter {
  import MyWorker._
  import context._

  def receive = idle

  var subTexts = Vector[List[String]]()
  val router = createWorkerRouter

  var workerGiven = 0
  var workReceived = 0
  var workers = Set[ActorRef]()

  def idle: Receive = {
    case StartJob(name,text) =>
      subTexts = text.grouped(10).toVector
      //定时给router发work消息
      val cancellable = context.system.scheduler.schedule(0 millis, 1000 millis,
        router,Work(name,self))
      setReceiveTimeout(60 seconds)
      //为了将结果传回receptionist，需要将sender反复传给其它状态
      become(working(name,sender(),cancellable))
  }

  //先不要管working状态下为什么会有这些参数，等到你感觉需要的时候
  //再添加也不迟
  def working(name: String,receptionist: ActorRef,
    cancellabe: Cancellable): Receive = {
    case Enlist(worker) =>
      watch(worker)
      workers = workers + worker
    case NextTask =>
      //这里面对于发送者是用sender还是通过将self传过来获得很随意
      if (subTexts.isEmpty) {
        sender() ! WorkDepleted
      } else {
        log.info("text not done, give one to worker")
        sender() ! Task(subTexts.head,self)
        workerGiven = workerGiven + 1
        subTexts = subTexts.tail
      }
  }
}